package com.antgroup.seckill.business.common.kafka;

import com.alibaba.fastjson.JSON;
import com.antgroup.seckill.business.common.constants.CommonConstant;
import com.antgroup.seckill.business.common.vo.response.GoodsVo;
import com.antgroup.seckill.business.goods.model.SeckillMessage;
import com.antgroup.seckill.business.goods.service.IGoodsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Objects;

/**
 * @author wangxy
 * @date 2024/9/6 23:52
 */
@Component
@Slf4j
public class KafkaConsumer {

    @Autowired
    private IGoodsService iGoodsService;

    @Autowired
    private RestHighLevelClient esClient;


    @KafkaListener(topics = {"goods-save"}, groupId = "saveGoods-group")
    public void onMessage(ConsumerRecord<?, ?> record) {

        System.out.println("发的消息为：" + record.value());

        log.info("处理新增消息: {}", record.value());

        String message = String.valueOf(record.value());

        SeckillMessage seckillMessage = JSON.parseObject(message, SeckillMessage.class);


        //查询商品信息 存入es
        GoodsVo goodsVo = iGoodsService.queryGoodsByGoodsId(seckillMessage.getGoogsId());

        if (Objects.isNull(goodsVo)) {
            return;
        }

        IndexRequest indexRequest = new IndexRequest(CommonConstant.GOODS_INDEX)
                .id(goodsVo.getId())
                .source(JSON.toJSONString(goodsVo), XContentType.JSON);

        try {
            esClient.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }


    }
}
